package com.ht.callback.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.ObjectMapper;

import cn.hutool.core.lang.UUID;
import lombok.extern.slf4j.Slf4j;

/**
 * kafka生产者
 * @author asq
 * @createTime 2024年12月7日 10:22:05
 */
@Slf4j
@Component
public class KafkaProducer {

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@Autowired
	private ObjectMapper objectMapper;

	/**
	 * 发送消息
	 */
	public <T> void sendMessage(String topic, T data) {
		sendMessage(topic, null, data);
	}

	/**
	 * 发送带类型的消息
	 */
	public <T> void sendMessage(String topic, String type, T data) {
		try {
			KafkaMessage<T> message = KafkaMessage.<T>builder().messageId(UUID.randomUUID().toString()).data(data)
					.type(type).timestamp(System.currentTimeMillis()).build();

			String jsonMessage = objectMapper.writeValueAsString(message);

			kafkaTemplate.send(topic, message.getMessageId(), jsonMessage);
			log.info("[kafka]发送消息成功，topic：{}，消息：{}", topic, jsonMessage);
		} catch (Exception e) {
			log.error("[kafka]消息发送异常", e);
			throw new RuntimeException("消息发送失败", e);
		}
	}
}